package com.lzwk.app.dwm;

/**
 * @Author: CC
 * @Date: 2022/1/5 17:57
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.lzwk.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;


public class UniqueVisitStatLog {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://10.2.16.4:4007/flink_113/checkpoints/ck");
        env.enableCheckpointing(600000L);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(100000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(60 * 1000);

        String groupId = "stat_log_attend_class_lh";
        String sourceTopic = "stat_log_mid_113";
        String sinkTopic = "topic2";
        String brokers = "127.0.0.1:9002";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId,brokers));

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

        // account_id+lecture_id进行key by
        // 1.根据account_id+lecture_id进行过滤，只输出第一条的数据

        KeyedStream<JSONObject, Tuple2<String, String>> KeyedStream = jsonObjDS.keyBy(new KeySelector<JSONObject, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(JSONObject value) throws Exception {
                return Tuple2.of(value.getString("account_id"), value.getString("lecture_id"));
            }
        });

        SingleOutputStreamOperator<JSONObject> uvDS = KeyedStream.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<Tuple2<String, String>> onlineState;
            private final List<String> actionList = Arrays.asList("classroom_stat_liveon", "classroom_stat_online");

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Tuple2<String, String>> valueStateDescriptor = new ValueStateDescriptor<Tuple2<String, String>>("attend_state", TypeInformation.of(new TypeHint<Tuple2<String, String>>() {
                }));
                //设置状态的超时时间以及更新时间的方式
                StateTtlConfig stateTtlConfig = new StateTtlConfig
                        .Builder(Time.hours(24))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build();
                valueStateDescriptor.enableTimeToLive(stateTtlConfig);
                onlineState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {
                Tuple2<String, String> curOnline = Tuple2.of(value.getString("account_id"), value.getString("lecture_id"));
                String action = value.getString("action");

                Tuple2<String, String> lastOnline = onlineState.value();
//                System.out.println("lastOnline>>>>>>>>:" + lastOnline);
//                System.out.println("curOnline >>>>>>>>:" + curOnline);

                if (lastOnline == null || !lastOnline.equals(curOnline)) {
                    if (actionList.contains(action)) {
                        onlineState.update(curOnline);
                        return true;
                    }
                }
                return false;
            }
        });
//
//        EnvironmentSettings settings = EnvironmentSettings.newInstance()
//                .useBlinkPlanner()
//                .inStreamingMode()
//                .build();
////
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
////        tableEnv.registerDataStream("data_table", uvDS);
//        Table table = tableEnv.scan("data_table");
//        table.printSchema();
//
//        Table table = tableEnv.fromDataStream(uvDS,"referer, remote_address, request, lecture_id, time_local, lts, message_id,remote_address,x_forwarded_for,playback_rate,learn_time,account_id,action,user_agent");
//        Table table = tableEnv.fromDataStream(uvDS,"f1, rowtime.rowtime, f0 as 'name'");
//
//        table.printSchema();
//
//        Table filterTable = table.select("'account_id=' + account_id,'lecture_id=' + lecture_id");
//        DataStream<Row> appendStream = tableEnv.toAppendStream(table, Row.class);
//        //并打印表数据
//
////        // 5. 将流转换成表
//       tableEnv.createTemporaryView("data_table",uvDS, "referer, remote_address, request, lecture_id, time_local, lts, message_id,remote_address,x_forwarded_for,playback_rate,learn_time,account_id,action,user_agent");
////
//////
//        Table resultSqlTable = tableEnv.sqlQuery("select * from data_table");
////
//        tableEnv.toRetractStream(resultSqlTable, Row.class).print();
////        appendStream.print();
////        2.转换成table进行lookup join关联维度，替换到播字段
//        uvDS.print();
        uvDS.map(JSONAware::toJSONString).addSink(MyKafkaUtil.getKafkaProducer(sinkTopic,brokers));

        env.execute("UniqueVisitStatLog");
    }
}

